package me.spring.cloud.common.components.rabbitmq.consumer;

import me.spring.cloud.common.components.rabbitmq.message.DLXMessage;
import me.spring.cloud.common.components.rabbitmq.message.IMessage;
import me.spring.cloud.common.components.rabbitmq.service.IMessageService;
import me.spring.cloud.common.components.rabbitmq.util.RabbitMqUtil;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * @description: rabbitmq消费者
 * @author: luffy
 * @create: 2019-01-18 14:02
 **/
@Slf4j
public abstract class AbstractRetryRabbitmqConsumer<T extends DLXMessage<? extends M>, M extends IMessage> {

  @Autowired
  private IMessageService iMessageService;

  /**
   * @Return void
   * @Author luffy
   * @Description 消息处理类
   * @Date 2019-05-30 10:00
   * @Param [message]
   **/
  public abstract void messageHandler(T message);

  /**
   * @param message
   * @Return void
   * @Author luffy
   * @Description 消息监听类，消费者
   * @Date 2019-05-30 08:44 s  * @Param [message]
   */
  public void consumerMessage(T message) {
    try {
      dealMessage(message.getContent());
    } catch (Exception e) {
      log.debug(">>>>> retry message:{}", message);
      try {
        this.retryMessage(message);
      } catch (Exception e1) {
        log.error(e.getMessage(), e);
        throw e;
      }
    }
  }

  /**
   * @param message
   * @Return void
   * @Author luffy
   * @Description 消费消息
   * @Date 2019-08-09 13:35
   */
  public abstract void dealMessage(M message);

  public abstract String getQueueName();

  public abstract String getExchangeName();

  private String getDeadQueueName() {
    return RabbitMqUtil.getDeadQueueName(getQueueName());
  }

  /**
   * @Return void
   * @Author luffy
   * @Description 处理重试
   * @Date 2019-05-29 17:40
   * @Param [message]
   **/
  private void retryMessage(T message) {
    // 不限制最大重试次数
    if (message.getMaxAttempts() <= 0) {
      // 0<最小延迟时间<最大延迟时间
      if (message.getMinInterval() > 0 && message.getMaxInterval() > message.getMinInterval()) {
        this.sengDLXMessage(message);
      } else {
        iMessageService.send(getQueueName(), message);
      }
    } else if (message.getMaxAttempts() >= message.getRetryNum()) {
      // 0<最小延迟时间<最大延迟时间
      if (message.getMinInterval() > 0 && message.getMaxInterval() > message.getMinInterval()) {
        // 当前重试延迟时间>最大延迟时间
        this.sengDLXMessage(message);
      } else {
        message.countRetryNum();
        iMessageService.send(getQueueName(), message);
      }
    } else if (message.getMaxAttempts() < message.getRetryNum()) {
      this.dealDeadMessage(message);
    }
  }

  private void sengDLXMessage(T message) {
    // 当前重试延迟时间（重试次数*时间增加倍数*最小延迟时间）>最大延迟时间
    if (message.getRetryNum() * message.getMultiple() * message.getMinInterval() > message.getMaxInterval()) {
      message.countRetryNum();
      this.sendDLXMessage(message, message.getMaxInterval());
    } else {
      message.countRetryNum();
      log.warn("该消息一直消费不了，请确认是否正常!!! 消息内容：{}", message);
      this.sendDLXMessage(message, message.getRetryNum() * message.getMultiple() * message.getMinInterval());
    }
  }

  /**
   * @Return void
   * @Author luffy
   * @Description 发送延迟消息到死信队列里面，等待重试
   * @Date 2019-05-29 18:01
   * @Param [message]
   **/
  private void sendDLXMessage(T message, long second) {
    iMessageService.send(getDeadQueueName(), getExchangeName(), message, second);
  }

  /**
   * @Author luffy
   * @Description 处理死信消息
   * @Param message
   * @Return void
   * @Date 2019-09-10 14:18
   */
  public void dealDeadMessage(T message) {
    log.warn(">>>>>>>>> 死信消息，内容：{}", message);
  }
}
